<?php

namespace extend\bridgecli;

use Swoole\Coroutine;
use Swoole\Coroutine\Channel;
use Swoole\Coroutine\Client;
use zhanshop\App;
use zhanshop\console\command\Server;
use zhanshop\Log;
use zhanshop\server\Request;
use zhanshop\server\WebHandle;

class BridgClient
{
    /**
     * 连接句柄
     * @var Client
     */
    protected $client;
    /**
     * @var Channel
     */
    public $bridgeChannel = null;
    public $cid = [];
    /**
     * 配置信息
     * @var array|mixed|string[]
     */
    public $config = [
        'type' => 1, // SWOOLE_SOCK_TCP
        'ip' => "127.0.0.1",
        'port' => "8201",
        'path' => '/v1/bridge', // 桥接的建立连接和吞吐消息的path前缀
        'servfd' => 0, // 当前桥接服务对应服务端的fd
        'backlog' => 7168, // 最大情况 单个连接7168 * 64K = 约会占用500M内存
        'worker_id' => 0,
        'timeout' => 3,
        'debug' => false,
    ];
    // 这玩意需要单独起进程 不然就是阻塞的
    public function __construct(int $workerId)
    {
        $config = App::config()->get('bridge');
        if($config) $this->config = array_merge($this->config, $config);
        $this->config['servfd'] = 0;
        $this->config['worker_id'] = $workerId;
        if($this->config['ip']){
            Coroutine::create(function () {
                for (;;) {
                    $this->listen();
                    sleep($this->config['timeout']);
                }
            });
        }
    }

    /**
     * 监听
     * @return void
     */
    protected function listen(){

        $client = new Client($this->config['type']);
        $this->config['options'] = $this->config['options'] ?? [];
        $client->set($this->config['options']);
        if (!$client->connect($this->config['ip'], intval($this->config['port']), 5))
        {
            $client->close();
            Log::errorLog(5, "桥服务tcp://".$this->config['ip'].":".$this->config['port'].'连接失败');
            return;
        }

        if($this->establish($client) == false) return;

        $this->eventLoop($client); // 开始事件循环 读取和发送消息

        Log::errorLog(3, "桥服务tcp://".$this->config['ip'].":".$this->config['port'].'连接被断开['.$this->config['servfd'].']');

        $client->close();
        $this->config['servfd'] = 0;

        foreach ($this->cid as $cid){
            Coroutine::cancel($cid);
        }
    }

    /**
     * 事件循环
     * @param $client
     * @return void
     */
    protected function eventLoop(Client $client){
        $backlog = intval($this->config['backlog']);
        $bridgeChannel = new Channel($backlog ? $backlog : 1024);
        $this->bridgeChannel = $bridgeChannel;

        $status = true;
        Coroutine::create(function () use ($client, &$bridgeChannel, &$status){
            while (true){
                $data = $bridgeChannel->pop(30);
                if($data){
                    $client->send($data."\r\n");
                }else if($bridgeChannel->errCode == SWOOLE_CHANNEL_TIMEOUT){
                    $sendStatus = $client->send("ping"."\r\n");
                    if($sendStatus == false){
                        swoole_error_log(5, "ping消息发送失败");
                        $status = false;
                        $client->close();
                        $bridgeChannel->close();
                        unset($bridgeChannel);
                        break;
                    }
                }
            }
        });

        $fullData = "";
        while ($status){
            $recv = $client->recv(10);
            if(strlen($recv) > 0){
                $fullData .= $recv;
                if(strpos($fullData, "\r\n") === false) continue;
                $arr = explode("\r\n", $fullData);
                for($i = 0; $i < count($arr) - 1; ++$i) {
                    $rowData = $arr[$i];
                    $fullData = substr($fullData, strlen($rowData) + 2);
                    $data = json_decode($rowData, true);
                    if($data == false) continue;
                    try {
                        Coroutine::create(function () use ($bridgeChannel, $data){
                            $this->message($bridgeChannel, $data);
                        });
                    }catch (\Throwable $e){
                        App::log()->push($recv->data.' => '.$e->getMessage().PHP_EOL.$e->getFile().':'.$e->getLine().PHP_EOL.$e->getTraceAsString(), 'BRIDGE-ERROR');
                    }
                }
            }else if($recv === ""){
                $status = false;
                $client->close();
                swoole_error_log(5, "读取服务器消息时发现服务器断开了".$client->errCode.':'.$client->errMsg);
                break;
            }else if($recv === false){
                if ($client->errCode !== SOCKET_ETIMEDOUT) {
                    $status = false;
                    $client->close();
                    swoole_error_log(5, "读取服务器消息时发生非超时异常".$client->errCode.':'.$client->errMsg);
                    break;
                }
            }else{
                $status = false;
                $client->close();
                swoole_error_log(5, "读取服务器消息时发生异常".$client->errCode.':'.$client->errMsg);
                break;
            }
        }
    }

    /**
     * 收到消息
     * @param array $data
     * @return void
     */
    protected function message(Channel $bridgeChannel, array $data){
        $request = \Swoole\Http\Request::create([]);
        $request->fd = $this->config['servfd'];
        $respData = [
            'uri' => $this->config['path'], // 响应消息给服务端
            'header' => [],
            'body' => [],
        ];
        if(isset($data['uri']) && isset($data['header']) && is_array($data['header'])){
            $request->server['request_uri'] = $data['uri'];
            $request->server['request_time'] = time();
            $request->server['request_method'] = 'TCP';
            $request->post = $data['body'] ?? [];
            foreach($data['header'] ?? [] as $k => $v){
                $request->header[$k] = $v;
            }
            $protocol = Server::TCP;
            $servRequest = new Request($protocol, $request);
            $result = App::make(WebHandle::class)->dispatchBridg('admin', $servRequest, $bridgeChannel);
            if($result !== true){
                Log::errorLog(5, $result);
                $toFd = intval($servRequest->getData("tofd", 0));
                if($toFd == false) return App::log()->push("tofd参数是空的", "BRIDG-ERROR");
                $this->send($toFd, $result, 'error'); // 原因里面还有协程么
            }
        }
    }

    /**
     * 建立连接
     * @param Client $client
     * @return void
     */
    public function establish(Client $client)
    {
        $bridgName = $this->config['name'] ?? ($_SERVER['HOSTNAME'] ?? "unknown");
        $respData = [
            'uri' => $this->config['path'].'.establish', // 响应消息给服务端
            'header' => [
                'code' => 0,
                'msg' => "ok",
            ],
            'body' => [
                'name' => $bridgName,
                'version' => PHP_OS.':'.PHP_VERSION,
                'token' => $this->config['token'] ?? ''
            ],
        ];
        $client->send(json_encode($respData)."\r\n");
        $fd = $client->recv();
        if(is_numeric($fd)){
            $fd = intval($fd);
            $this->config['servfd'] = $fd;
            Log::errorLog(3, "桥服务tcp://".$this->config['ip'].":".$this->config['port'].'连接建立成功['.$fd.']');
            return true;
        }else{
            $fd = str_replace(["\r", "\n"], '', $fd);
            Log::errorLog(5, "桥服务tcp://".$this->config['ip'].":".$this->config['port'].'连接建立失败['.$fd.']');
        }
        return false;
    }

    /**
     * 给桥接fd发送结果消息
     * @param int $tofd
     * @param mixed $body
     * @param string $event
     * @return void
     */
    public function send(int $tofd, mixed $body, string $event = 'result'){
        $respData = [
            'uri' => $this->config['path'].'.message/'.$tofd.'/'.$event, // 响应消息给服务端
            'header' => [
                'code' => 0,
                'msg' => "ok",
            ],
            'body' => base64_encode($body),
        ];
        return $this->toPush(json_encode($respData));
    }

    /**
     * 给桥接fd发送错误消息
     * @param int $tofd
     * @param mixed $body
     * @return void
     */
    public function error(int $tofd, mixed $body){
        $respData = [
            'uri' => $this->config['path'].'.message/'.$tofd.'/error', // 响应消息给服务端
            'header' => [
                'code' => 500,
                'msg' => "error",
            ],
            'body' => $body,
        ];
        return $this->toPush(json_encode($respData));
    }

    /**
     * 给桥接服务端方法发送消息
     * @param string $tofd
     * @param mixed $body
     * @return void
     */
    public function method(string $method, mixed $body){
        $respData = [
            'uri' => $this->config['path'].'.'.$method, // 响应消息给服务端
            'header' => [
                'code' => 0,
                'msg' => "ok",
            ],
            'body' => $body,
        ];
        return $this->toPush(json_encode($respData));
    }

    /**
     * 关闭掉指定桥接的fd
     * @param int $tofd
     * @return void
     */
    public function close(int $tofd)
    {
        $respData = [
            'uri' => $this->config['path'].'.message/'.$tofd.'/close', // 响应消息给服务端
            'header' => [
                'code' => 0,
                'msg' => "ok",
            ],
            'body' => null,
        ];

        return $this->toPush(json_encode($respData));
    }

    public function toPush(string $data)
    {
        if(!$this->bridgeChannel->push($data, 0.1)){
            if($this->bridgeChannel->errCode == SWOOLE_CHANNEL_CLOSED){
                // 通道关闭了
                swoole_error_log(5, "桥连接推送消息的时候发现通道已关闭");
            }else if($this->bridgeChannel->errCode == SWOOLE_CHANNEL_TIMEOUT){
                swoole_error_log(5, "桥连接推送消息通道写入超时");
            }
            return false;
        }
        return true;
    }
}
